package com.kingsoft.dc.khaos.module.spark.util


import com.kingsoft.dc.khaos.module.spark.constants.CosConstants
import com.kingsoft.dc.khaos.module.spark.model.cos.CosAccessConfig
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path

/**
  * Created by zoujing on 19/4/1.
  */
object HadoopCosUtils {
  val cosnBlockSize = 256 * 1024 * 1024
  val cosnBufferSize = 256 * 2 * 1024 * 1024
  val localBlockSize = 256 * 1024 * 1024

  val PATH_DELIMITER = Path.SEPARATOR

  //  def appendCosHadoopConfigs(hadoopConf: Configuration, cosAccessConfig: CosAccessConfig):Configuration = {
  //    val fileSystem = "cosn://" + cosAccessConfig.getBucket
  //    hadoopConf.set("hadoop.mapred.output.compress", "false")
  //    hadoopConf.set("fs.defaultFS", fileSystem)
  //    hadoopConf.set("fs.cosn.credentials.provider", "org.apache.hadoop.fs.auth.SimpleCredentialProvider")
  //    hadoopConf.set("fs.cosn.impl", "org.apache.hadoop.fs.CosFileSystem")
  //    hadoopConf.set("fs.AbstractFileSystem.cosn.impl", "org.apache.hadoop.fs.CosN")
  //
  //    if (cosAccessConfig.getWorkingDir != null) {
  //      hadoopConf.set("fs.cosn.tmp.dir", fileSystem + cosAccessConfig.getWorkingDir);
  //    } else {
  //      hadoopConf.set("fs.cosn.tmp.dir", fileSystem + "/di/tmp/");
  //    }
  //
  //    if (cosAccessConfig.getCosnBufferSize != null) {
  //      hadoopConf.set("fs.cosn.buffer.size", cosAccessConfig.getCosnBufferSize)
  //    } else {
  //      hadoopConf.set("fs.cosn.buffer.size", String.valueOf(cosnBufferSize))
  //    }
  //    if (cosAccessConfig.getCosnBlockSize != null) {
  //      hadoopConf.set("fs.cosn.block.size", cosAccessConfig.getCosnBlockSize)
  //    } else {
  //      hadoopConf.set("fs.cosn.block.size", String.valueOf(cosnBlockSize))
  //    }
  //    if (cosAccessConfig.getLocalBlockSize != null) {
  //      hadoopConf.set("fs.local.block.size", cosAccessConfig.getLocalBlockSize)
  //      hadoopConf.set("dfs.blocksize", cosAccessConfig.getLocalBlockSize)
  //    } else {
  //      hadoopConf.set("fs.local.block.size", String.valueOf(localBlockSize))
  //      hadoopConf.set("dfs.blocksize", String.valueOf(localBlockSize))
  //    }
  //    hadoopConf.set("fs.cosn.maxRetries", "3")
  //    hadoopConf.set("fs.cosn.retry.interval.seconds", "3")
  //    hadoopConf.set("fs.cosn.userinfo.secretId", cosAccessConfig.getAccessKey)
  //    hadoopConf.set("fs.cosn.userinfo.secretKey", cosAccessConfig.getSecretKey)
  //    hadoopConf.set("fs.cosn.userinfo.region", cosAccessConfig.getRegion)
  //    hadoopConf.set("fs.cosn.userinfo.endpoint_suffix", cosAccessConfig.getEndPoint)
  //    hadoopConf.set("hadoop.fs.cos.userinfo.appid", cosAccessConfig.getAppId)
  //    hadoopConf
  //  }

  def keyToPath(key: String): String = {
    if (!key.startsWith(PATH_DELIMITER)) {
      s"/${key}"
    }else{
      key
    }
  }


  /**
    * 加载默认配置
    * @param hadoopConf
    * @return
    */
  def loadDefault(hadoopConf: Configuration): Configuration = {
    hadoopConf.addResource(this.getClass.getClassLoader.getResourceAsStream(CosConstants.COS_HADOOP_CONFIG_FILE))
    hadoopConf
  }

  def appendCosHadoopConfigs(hadoopConf: Configuration, cosAccessConfig: CosAccessConfig): Configuration = {
    val hadoopConfNew = loadDefault(hadoopConf)
    val fsName = "cosn://" + cosAccessConfig.getBucket
    hadoopConfNew.set(CosConstants.DEFAULT_FS, fsName)
    if (cosAccessConfig.getWorkingDir != null) {
      hadoopConfNew.set(CosConstants.FS_COSN_TMP_DIR, cosAccessConfig.getWorkingDir)
    }
//    if (hadoopConfNew.get(CosConstants.FS_COSN_TMP_DIR).startsWith(PATH_DELIMITER)) {
//      hadoopConfNew.set(CosConstants.FS_COSN_TMP_DIR, fsName + hadoopConfNew.get(CosConstants.FS_COSN_TMP_DIR))
//    }
    hadoopConfNew.set(CosConstants.FS_COSN_USERINFO_SECRETID, cosAccessConfig.getAccessKey)
    hadoopConfNew.set(CosConstants.FS_COSN_USERINFO_SECRETKEY, cosAccessConfig.getSecretKey)
    hadoopConfNew.set(CosConstants.FS_COSN_USERINFO_REGION, cosAccessConfig.getRegion)
    hadoopConfNew.set(CosConstants.FS_COSN_USERINFO_ENDPOINT_SUFFIX, cosAccessConfig.getEndPoint)
    hadoopConfNew.set(CosConstants.FS_COSN_BUCKET_ENDPOINT_SUFFIX, cosAccessConfig.getEndPoint)  //新版参数
    hadoopConfNew.set(CosConstants.FS_COSN_BUCKET_REGION, cosAccessConfig.getRegion)  //新版参数

//    printConf(hadoopConfNew)
    hadoopConfNew
  }


  def printConf(hadoopConf: Configuration) = {
    println("---------------")
    
    val iters = hadoopConf.iterator()
    while (iters.hasNext) {
      val iter = iters.next()
      val key = iter.getKey
      val value = iter.getValue
      println(key+"==>"+value)
    }
    println("---------------")
  }
}
